package com.hh.bss.xthread;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.LinkedList;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.hh.bss.util.DateHelper;




public class XThreadPoolImpl implements XThreadPool{
	private static final Log logger = LogFactory.getLog(Worker.class);
	// 工人
	class Worker extends Thread {
		private long activeTime;
		private boolean active;
		private boolean shutdown;
		private XThreadRunnable xrunnable;
		public void run(){
			while(!shutdown){
				// 工人们都围绕着流水线转，从流水线上接工作
				synchronized(queueList){
					while(queueList.isEmpty()){
						try{
							queueList.wait();
						}catch(InterruptedException interruptedexception){
							return;
						}
					}
					xrunnable = (XThreadRunnable) queueList.removeFirst();
				}
				active = true;
				activeTime = System.currentTimeMillis();
				try{
					xrunnable.run();
					xrunnable.stop();// 回收资源
				}catch(RuntimeException e){
					System.err.println(e);
				}
				active = false;
				activeTime = System.currentTimeMillis();
			}
		}

		public XThreadRunnable getXRunnable(){
			return xrunnable;
		}

		public long getActiveTime(){
			return activeTime;
		}

		public boolean isActive(){
			return active;
		}

		public void setShutdown(boolean shutdown){
			this.shutdown = shutdown;
		}

		public String toString(){
			StringBuffer buffer = new StringBuffer();
			buffer.append("[name=" + getName());
			buffer.append(";active=" + isActive());
			buffer.append(";activeTime=" + DateHelper.toString(new Date(getActiveTime()), "yyyy-MM-dd HH:mm:ss"));
			if(xrunnable != null) buffer.append(";xrunnable=yes");
			buffer.append("]");
			return buffer.toString();
		}

		public Worker(){
			super("worker:" + DateHelper.toString(new Date(System.currentTimeMillis()), "yyyyMMddHHmmss"));
			active = false;
			shutdown = false;
			xrunnable = null;
			setDaemon(true);
		}
	}
	// 监控器
	class Monitor extends Thread {
		public void run(){
			do{
				try{
					Thread.sleep(monitorInterval);// access1
					checkWorkers();
				}catch(Exception e){
					logger.error("error:" + e.getMessage(), e);
				}
			}while(true);
		}
		// 资源回收[回收空闲Worker]和强制停止执行超时的Worker
		public void checkWorkers(){
			int activeCount = 0;
			Worker worker = null;
			synchronized(queueList){
				for (Iterator ir = workerList.iterator(); ir.hasNext();){
					worker = (Worker) ir.next();
					if(worker.isActive()){
						if(execTimeout != 0L && System.currentTimeMillis() - worker.getActiveTime() > execTimeout){
							try{
								worker.getXRunnable().stop();// 回收资源
								worker.setShutdown(true);
							}catch(Exception e){
								System.err.println(e);
							}
							if(debug){
								System.err.println("[" + worker + "]: worker timeout for execution");
							}
							worker = null;
							ir.remove();// 解聘掉工作超时的工人[可能存在问题]，此工作就丢掉了
						}else{
							activeCount++;// 处于工作中的
						}
					}else if(idleTimeout != 0L && workerList.size() > minWorkerCount
							&& System.currentTimeMillis() - worker.getActiveTime() > idleTimeout){
						worker.interrupt();
						if(debug){
							System.err.println("[" + worker + "]: worker timeout for idle");
						}
						worker = null;
						ir.remove();// 工人清闲太久了，节省资源，先解聘
					}
				}
				// 如果当前有工作，但已经无空闲的工人且还没有用完招聘名额,招聘新的工人
				if((workerList.size() - activeCount) <= 0 && workerList.size() < maxWorkerCount && queueList.size() > 0){
					worker = new Worker();
					workerList.add(worker);
					worker.start();
					if(debug){
						System.err.println("[" + name + "] new worker created " + workerList.size() + " " + queueList.size());
					}
				}

			}
			if(debug){
				System.err.println("[" + name + "] "
						+ DateHelper.toString(new Date(System.currentTimeMillis()), "yyyy-MM-dd HH:mm:ss") + ":active="
						+ activeCount + ",queue=" + queueList.size() + ",worker=" + workerList.size());
			}
		}

		public Monitor(){
			super();
			setDaemon(true);
		}
	}

	private final ArrayList workerList = new ArrayList();
	private final LinkedList queueList = new LinkedList();
	private int maxWorkerCount;
	private int minWorkerCount;
	private int maxQueueSize;
	private long execTimeout;
	private long idleTimeout;

	public XThreadPoolImpl(){
		debug = false;
		name = "XThreadPool";
		maxWorkerCount = 20;
		minWorkerCount = 20;
		maxQueueSize = 100;
		execTimeout = 1000 * 650;// 默认150秒超时
		idleTimeout = 1000 * 300;// 默认30秒回收
		monitorInterval = 5000;// 默认5秒监控一次
	}

	public void initialize(){
		logger.info("maxWorkerCount=" + maxWorkerCount + ",minWorkerCount=" + minWorkerCount + ",execTimeout="
				+ execTimeout + ",idleTimeout=" + idleTimeout + ",monitorInterval=" + monitorInterval);
		for (int i = 0; i < minWorkerCount; i++){
			Worker worker = new Worker();
			workerList.add(worker);
			worker.start();
		}
		(new Monitor()).start();
	}

	public void execute(XThreadRunnable xrunner){
		// 暂止流水线
		synchronized(queueList){
			// 将工作放到流水线上
			queueList.addLast(xrunner);
			// 当前在忙的工人数量
			int activeCount = 0;
			for (Iterator ir = workerList.iterator(); ir.hasNext();)//
			{
				Worker worker = (Worker) ir.next();
				if(worker.isActive()){
					activeCount++;// 处于工作中的
				}
			}
			// 如果已经无空闲的工人且还没有用完招聘名额,招聘新的工人
			if((workerList.size() - activeCount) <= 0 && workerList.size() < maxWorkerCount){
				Worker worker = new Worker();
				workerList.add(worker);
				worker.start();// 马上入职(在流水线上接工作)
				if(debug){
					System.err.println("[" + name + "] : new worker created " + workerList.size() + " " + queueList.size());
				}
			}
			// 开启流水线，让有准备的工人可以继续接工作
			queueList.notify();
		}
	}

	public String toString(){
		Worker worker = null;
		StringBuffer buffer = new StringBuffer();
		buffer.append(queueList.size()).append(" queued\t");
		buffer.append(workerList.size()).append(" running\r\n");
		synchronized(queueList){
			try{
				for (Iterator ir = workerList.iterator(); ir.hasNext(); buffer.append(worker.toString()).append("\r\n")){
					worker = (Worker) ir.next();
				}

			}catch(Exception e){
				logger.info("error:" + e.getMessage(), e);
			}
		}
		return buffer.toString();
	}

	public int getWorkerCount(){
		return workerList.size();
	}

	public int getQueueSize(){
		return queueList.size();
	}
	public void setMaxWorkerCount(int maxWorkerCount){
		this.maxWorkerCount = maxWorkerCount;
	}
	public int getMaxWorkerCount(){
		return maxWorkerCount;
	}
	public void setMinWorkerCount(int minWorkerCount){
		this.minWorkerCount = minWorkerCount;
	}
	public int getMinWorkerCount(){
		return minWorkerCount;
	}

	public void setMaxQueueSize(int maxQueueSize){
		this.maxQueueSize = maxQueueSize;
	}
	public int getMaxQueueSize(){
		return maxQueueSize;
	}
	public void setExecTimeout(long execTimeout){
		this.execTimeout = execTimeout;
	}

	public long getExecTimeout(){
		return execTimeout;
	}

	public void setIdleTimeout(long idleTimeout){
		this.idleTimeout = idleTimeout;
	}
	public long getIdleTimeout(){
		return idleTimeout;
	}

	public boolean overLoaded(){
		return queueList.size() >= maxQueueSize;
	}

	private String name = null;
	public void setName(String name){
		this.name = name;
	}
	public String getName(){
		return this.name;
	}
	private long monitorInterval = 0L;
	public void setMonitorInterval(long monitorInterval){
		this.monitorInterval = monitorInterval;
	}
	public long getMonitorInterval(){
		return this.monitorInterval;
	}
	private boolean debug = false;
	public void setDebug(boolean debug){
		this.debug = debug;
	}
	public boolean getDebug(){
		return this.debug;
	}
}
